package com.potenza.sendorder.controller;

import com.alibaba.fastjson.JSON;
import com.potenza.sendorder.domain.Order;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Arrays;
import java.util.Iterator;

@RestController
    @RequestMapping("sendorder")
public class SendOrderController {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    KafkaConsumer<String, String> kafkaConsumer;

    @Autowired
    RedisTemplate redisTemplate;

    @GetMapping("send")
    public Order send(){
        Order order = new Order();
        order.setLocation("116.298964");
        order.setLatitude("40.043093");
//        ProducerRecord producerRecord = new ProducerRecord("add", "location", JSON.toJSONString(order));
        kafkaTemplate.send("add", "location", JSON.toJSONString(order));
        return order;
    }




    /**
     * 监听kafka消息 拉取消息
     */
    @PostConstruct// Bean创建完成后，会调用PostConstruct 指定的方法
    public void init(){
        System.out.println("ConsumerPoll对象创建完后，执行的初始化代码");
        kafkaConsumer.subscribe(Arrays.asList("add"));

        Thread thread = new Thread(()->{
            while(true){
                poll();
            }
        });

        thread.start();
        System.out.println("Poll Thread:" + thread.getName());
    }



    public void poll() {
        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));//主动拉取
        Iterator<ConsumerRecord<String, String>> iterator = consumerRecords.iterator();
        while (iterator.hasNext()) {
            ConsumerRecord<String, String> consumerRecord = iterator.next();
            System.out.println("5:" + Thread.currentThread().getName());
            System.out.println("5：接收到信息(string)，key：" + consumerRecord.key() +
                    " value:" + consumerRecord.value());

//            redisTemplate.opsForGeo().add("SJJWD",consumerRecord.value(),"司机");
            //确保业务执行正确在确认
            // 手动提交 offset 位置
            GeoOperations geoOperations = redisTemplate.opsForGeo();
            for (int i = 1; i <= 10; i++) {
                String member = "司机"+i;
                 geoOperations.geoDist("SJJWD", "司机0", member, RedisGeoCommands.DistanceUnit.METERS).var
//                geoOperations.geoDist("GET_DISTANCE", consumerRecord.value(), member, GeoUnit.KM);
                System.out.println("Distance between your location and member" + i + ": " + get_distance.getValue() + " kilometers");
            }

            kafkaConsumer.commitSync();//手动提交
        }
    }


    //    @GetMapping(value = {
//            "sendMessageObject/{userName}/{price}/{partition}",
//            "sendMessageObject/{userName}/{price}"
//    })
//    public Order sendMessageObject(@PathVariable String userName,
//                                   @PathVariable Integer price,
//                                   @PathVariable(required = false) Integer partition) {
//
//        Order order = new Order();
//        order.setId(orderId++);
////        order.setUserName(userName);
////        order.setName("订单:" + orderId);
////        order.setPrice(price);
//        System.out.println("待处理的订单已纳入处理队列（kafka），id："+orderId);
//
//
//        /**
//         * topic 主题
//         * key
//         * value
//         * 没有指定分片：策略
//         */
//        if(partition == null) {
//            kafkaTemplate.send("add", userName, order);//使用send方法发送消息，需要传入topic名称
//        }
//        kafkaTemplate.send("add", partition,userName,order);
//        /*
//        kafkaTemplate.send("my-2109A-topic", 0,userName,order);
//        kafkaTemplate.send("my-2012A-topic", 1,userName,order);
//        */
//
//        return order;
//    }

}
